// Copyright 2008 and onwards Matt Burkhart, Mike Chan.
//
// This program is free software; you can redistribute it and/or modify it under
// the terms of the GNU General Public License as published by the Free Software
// Foundation; version 2 of the License.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
// FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
// details.

#include <assert.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/statvfs.h>
#include <sys/wait.h>
#include <unistd.h>
#include <algorithm>
#include <fstream>
#include <set>
#include <stack>

#include "switchfs.h"

#include "path_utils.h"
#include "source.h"

using namespace std;

#define LOG(format, args...) logger_->Log(format, ## args)

#ifdef VERBOSE
#define DEBUG(X) LOG(X)
#else
#define DEBUG(X)
#endif

#define source_map(source)  \
  for (vector<Source>::iterator source = sources_.begin();  \
       source != sources_.end(); ++source)

SwitchFS::SwitchFS(Logger* logger) : logger_(logger) {
  assert(logger_ != NULL);
  pthread_rwlock_init(&sources_lock_, NULL);
}

void SwitchFS::AddSource(const string& location) {
  pthread_rwlock_wrlock(&sources_lock_);
  assert(find(sources_.begin(), sources_.end(), location) == sources_.end());
  sources_.push_back(Source(location));
  pthread_rwlock_unlock(&sources_lock_);
}

int SwitchFS::Check(const string& virtual_path, bool balance) {
  set<string> unique_files;
  set<string> unique_directories;

  source_map(source) {
    string physical_path = Join(*source, virtual_path);
    DIR* dp = ::opendir(physical_path.c_str());
    if (!dp) {
      // Magic number 493 is taken from various SwitchFS logging. Oddly, number
      // does not match stat --format=%f foo which appears to yield 0x41ed.
      // Investigation needed.
      LOG("DIR not found. Creating: %s", physical_path.c_str());
      if (::mkdir(physical_path.c_str(), 493) == -1) {
        LOG("Error creating dir: %s (%d)", strerror(errno), errno);
        return -errno;
      }
      continue;
    }

    // Track all unique file types that are DT_DIR (directories) for recursion
    // and DT_REG (regular files) for balancing.
    struct dirent *de;
    while ((de = ::readdir(dp))) {
      string name = de->d_name;
      // Only insert unique dir names. Ignore "." and ".." dirs.
      if (de->d_type == DT_DIR && name != "." && name != "..") {
        unique_directories.insert(de->d_name);
      } else if (balance && de->d_type == DT_REG) {
        unique_files.insert(de->d_name);
      }
    }
    ::closedir(dp);
  }

  for (set<string>::const_iterator file = unique_files.begin();
       file != unique_files.end(); ++file) {
    string virtual_file_path = Join(virtual_path, *file);
    int error = ReplicateFile(virtual_file_path);
    if (error) return error;
    error = BalanceFile(virtual_file_path);
    if (error) return error;
  }
  for (set<string>::const_iterator directory = unique_directories.begin();
       directory != unique_directories.end(); ++directory) {
    int error = Check(Join(virtual_path, *directory), balance);
    if (error) return error;
  }
  return 0;
}

int SwitchFS::ReplicateFile(const string& virtual_path) {
  size_t target_replica_count = GetDesiredReplication(virtual_path);
  size_t current_replica_count = GetFileSources(virtual_path).size();

  if (current_replica_count < target_replica_count) {
    LOG("File \"%s\" is under-replicated; %d versus %d. Creating replicas:",
        virtual_path.c_str(), current_replica_count, target_replica_count);
    size_t difference = target_replica_count - current_replica_count;

    // Note that the call to AllocateSources(...) invalidates the Source pointer
    // addresses (since it sorts the sources vector) so it must be called before
    // we get the list of current file sources.
    vector<Source*> optimal_sources = AllocateSources(sources_.size());
    vector<Source*> current_sources = GetFileSources(virtual_path);
    sort(optimal_sources.begin(), optimal_sources.end());
    sort(current_sources.begin(), current_sources.end());
    vector<Source*> target_sources(optimal_sources.size());
    vector<Source*>::iterator target_sources_end =
        set_difference(optimal_sources.begin(), optimal_sources.end(),
                       current_sources.begin(), current_sources.end(),
                       target_sources.begin());
    sort(target_sources.begin(), target_sources_end, SourceComparitor());

    // Starting with the drive with the highest allocation priority, copy new
    // physical replicas until the desired replication level is reached.
    for (size_t replica = 0; replica < difference; ++replica) {
      string old_physical_path = Join(*current_sources.front(), virtual_path);
      string new_physical_path = Join(*target_sources[replica], virtual_path);
      int error = CopyPhysicalFile(old_physical_path, new_physical_path);
      if (error) return error;
    }
  }

  if (current_replica_count > target_replica_count) {
    LOG("File \"%s\" is over-replicated; %d versus %d. Destroying replicas:",
        virtual_path.c_str(), current_replica_count, target_replica_count);
    size_t difference = current_replica_count - target_replica_count;
    vector<Source*> current_sources = GetFileSources(virtual_path);
    sort(current_sources.begin(), current_sources.end(), SourceComparitor());

    // Starting with the file on the most over crowded drive, delete physical
    // replicas until the desired replication level is reached.
    for (size_t replica = 0; replica < difference; ++replica) {
      string physical_path = Join(*current_sources.back(), virtual_path);
      current_sources.pop_back();
      int error = RemovePhysicalFile(physical_path);
      if (error) return error;
    }
  }
  return 0;
}

int SwitchFS::BalanceFile(const string& virtual_path) {
  // The balance threshold determines the minimum amount by which the free space
  // must differ between current and desired sources for a balance move to take
  // place.
  const unsigned long long kBalanceThreshold = 1024 * 1024 * 1024;  // 1 GB.

  // Note that the call to AllocateSources(...) invalidates the Source pointer
  // addresses (since it sorts the sources vector) so it must be called before
  // we get the list of current file sources.
  size_t replication_count = GetFileSources(virtual_path).size();
  vector<Source*> optimal_sources = AllocateSources(replication_count);
  vector<Source*> current_sources = GetFileSources(virtual_path);

  // Given the list of current file sources and the optimal file sources,
  // determine which of the optimal sources to move to and which current sources
  // to move from based off of which has the most available free space.
  sort(current_sources.begin(), current_sources.end());
  sort(optimal_sources.begin(), optimal_sources.end());
  vector<Source*> target_sources(current_sources.size());
  vector<Source*>::iterator target_sources_end =
      set_difference(optimal_sources.begin(), optimal_sources.end(),
                     current_sources.begin(), current_sources.end(),
                     target_sources.begin());
  sort(current_sources.begin(), current_sources.end(), SourceComparitor());
  sort(target_sources.begin(), target_sources_end, SourceComparitor());

  // For each target source *starting with the least allocated* move a file
  // replica from the *most over-allocated* source. Stop if we have moved to all
  // targets or even the most over-allocated source is in a similar (enough as
  // defined by kBalanceThreshold) situation.
  for (vector<Source*>::const_iterator target_source = target_sources.begin();
       target_source != target_sources_end; ++target_source) {
    Source* current_source = current_sources.back();
    current_sources.pop_back();
    if ((*target_source)->GetFreeSpace() - current_source->GetFreeSpace() <
        kBalanceThreshold) {
      break;
    }

    string old_physical_path = Join(*current_source, virtual_path);
    string new_physical_path = Join(**target_source, virtual_path);
    int error = MovePhysicalFile(old_physical_path, new_physical_path);
    if (error) return error;
  }
  return 0;
}

vector<Source*> SwitchFS::AllocateSources(size_t count) {
  // This method decides on which sources we should place new files. We choose
  // "count" sources. count must be less or equal to the number of sources we
  // know of and greater than 0.
  pthread_rwlock_wrlock(&sources_lock_);
  assert(count <= sources_.size());
  assert(count > 0);

  // Skip any sorting computation if we do not have any choice where to
  // replicate since we have to replicated on all sources. This implementation
  // uses the operators defined in the source class to dictate allocation
  // ordering.
  if (count != sources_.size()) {
    sort(sources_.begin(), sources_.end());
  }

  // Return the correct, desired amount of sources to the caller.
  vector<Source*> result;
  for (vector<Source>::iterator source = sources_.begin();
       result.size() < count; ++source) {
    result.push_back(&(*source));
  }
  pthread_rwlock_unlock(&sources_lock_);

  assert(result.size() == count);
  return result;
}

vector<Source*> SwitchFS::GetFileSources(const string& virtual_path) {
  vector<Source*> result;

  pthread_rwlock_rdlock(&sources_lock_);
  source_map(source) {
    string file_path = Join(*source, virtual_path);
    if (::access(file_path.c_str(), F_OK) == 0) {
      result.push_back(&(*source));
    }
  }
  pthread_rwlock_unlock(&sources_lock_);
  return result;
}

Source* SwitchFS::GetFileSource(const string& virtual_path) {
  // All of them should be the same, so just return one.
  vector<Source*> sources = GetFileSources(virtual_path);

  // File exists.
  if (sources.size() > 0) return *sources.begin();

  // File does not yet exist. We can return an hypothetical path for methods
  // which expect a result regardless of file existence.
  assert(sources_.size() >= 1);
  return &sources_[0];
}

vector<string> SwitchFS::GetFilePaths(const string& virtual_path) {
  vector<Source*> sources = GetFileSources(virtual_path);

  vector<string> result;
  for (vector<Source*>::iterator source = sources.begin();
       source != sources.end(); ++source) {
    result.push_back(Join(**source, virtual_path));
  }
  return result;
}

string SwitchFS::GetFilePath(const string& virtual_path) {
  return Join(*GetFileSource(virtual_path), virtual_path);
}

size_t SwitchFS::GetDesiredReplication(const string& virtual_path) {
  // If this is a replication directive, we need to put it on all drives.
  if (Basename(virtual_path) == GetDirectiveFilename()) {
    pthread_rwlock_rdlock(&sources_lock_);
    size_t source_count = sources_.size();
    pthread_rwlock_unlock(&sources_lock_);
    return source_count;
  }

  // If not, find the corresponding replication directive file. We assume that
  // the replication directions are stored on each source so it doesn't matter
  // which source path we crawl.
  size_t desired_replication = GetDefaultReplicationLevel();
  pthread_rwlock_rdlock(&sources_lock_);
  string source_path = *sources_.begin();
  for (string path = StripBasename(Join(source_path, virtual_path));
       path.size() >= source_path.size(); path = StripBasename(path)) {
    string directive_path = Join(path, GetDirectiveFilename());

    // Check if the file exists. If so, open if an read out the replication
    // level. If the file is malformed, all we can do is print out a warning and
    // ignore the file.
    DEBUG("Looking for directive file: " << directive_path << endl);
    if (::access(directive_path.c_str(), F_OK) == 0) {
      ifstream directive_file(directive_path.c_str());
      int replication_level;
      if (directive_file >> replication_level) {
        assert(replication_level > 0);
        desired_replication = static_cast<size_t>(replication_level);
        break;
      }
      DEBUG("Warning: Malformed file. Could not read replication.");
    }
  }
  pthread_rwlock_unlock(&sources_lock_);

  assert(desired_replication > 0);
  return desired_replication;
}

int SwitchFS::MovePhysicalFile(const string& physical_path_source,
                               const string& physical_path_dest) {
  LOG("Moving \"%s\" to \"%s\".",
      physical_path_source.c_str(), physical_path_dest.c_str());
  const char* const args [] = {
    "--", physical_path_source.c_str(), physical_path_dest.c_str(), NULL };
  return ExecCommand("mv", args);
}

int SwitchFS::CopyPhysicalFile(const string& physical_path_source,
                               const string& physical_path_dest) {
  LOG("Copying \"%s\" to \"%s\".",
      physical_path_source.c_str(), physical_path_dest.c_str());
  const char* const args [] = {
    "--", physical_path_source.c_str(), physical_path_dest.c_str(), NULL };
  return ExecCommand("cp", args);
}

int SwitchFS::RemovePhysicalFile(const string& physical_path) {
  LOG("Removing \"%s\".", physical_path.c_str());
  const char* const args [] = { "--", physical_path.c_str(), NULL };
  return ExecCommand("rm", args);
}

int SwitchFS::ExecCommand(const char* cmd, const char* const* args) {
  pid_t pid = fork();
  if (pid == 0) {  // Child process.
    execvp(cmd, (char* const*)args);
    exit(1);
  } else {  // Parent process.
    int status;
    waitpid(pid, &status, 0);
    return WIFEXITED(status) && !WEXITSTATUS(status) ? 0 : -1;
  }
}

Logger& SwitchFS::logger() {
  assert(logger_ != NULL);
  return *logger_;
}

////
// SwitchFS Settings

string SwitchFS::GetDirectiveFilename() {
  return ".backup";
}

size_t SwitchFS::GetDefaultReplicationLevel() {
  return 1;
}

string SwitchFS::GetDeletedPrefix() {
  return ".trash-switchfs";
}

unsigned long long SwitchFS::GetBytesReserved() {
  return 1024 * 1024;  // 1 MB
}

////
// FUSE Interface

int SwitchFS::statfs(const char* path, struct statvfs* buffer) {
  // We need to return aggregate data for this file system. In most cases, this
  // is a product of the source file systems we are mapping to. It would be nice
  // to be able to cache this information.

  buffer->f_bsize = 0;
  pthread_rwlock_rdlock(&sources_lock_);
  source_map(source) {
    // Start off using the data from our first source with our specific flags.
    // We can then start adding statistics on top of that.
    if (buffer->f_bsize == 0) {
      ::statvfs(*source, buffer);
      buffer->f_fsid = 0x1337;  // Is this already used?
    }

    // Aggregate statistics from further sources.
    else {
      struct statvfs source_buffer;
      ::statvfs(*source, &source_buffer);

      // TODO: How to incorporate different block sizes and fragment sizes:
      // buffer->f_bsize, buffer->f_frsize.
      buffer->f_blocks += source_buffer.f_blocks;
      buffer->f_bfree += source_buffer.f_bfree;
      buffer->f_bavail += source_buffer.f_bavail;
      buffer->f_files += source_buffer.f_files;
      buffer->f_ffree += source_buffer.f_ffree;
      buffer->f_favail += source_buffer.f_favail;
      buffer->f_flag |= source_buffer.f_flag;
      buffer->f_namemax = min<unsigned long>(buffer->f_namemax,
					     source_buffer.f_namemax);
    }
  }
  pthread_rwlock_unlock(&sources_lock_);
  return 0;
}

int SwitchFS::access(const char* path, int mode) {
  return ::access(GetFilePath(path).c_str(), mode);
}

int SwitchFS::getattr(const char* path, struct stat* attributes) {
  if (::stat(GetFilePath(path).c_str(), attributes)) {
    return -errno;
  }
  return 0;
}

int SwitchFS::chmod(const char* path, mode_t mode) {
  vector<string> paths = GetFilePaths(path);
  for (vector<string>::const_iterator physical_path = paths.begin();
       physical_path != paths.end(); ++physical_path) {
    if (::chmod(physical_path->c_str(), mode)) {
      return -errno;
    }
  }
  return 0;
}

int SwitchFS::chown(const char* path, uid_t uid, gid_t gid) {
  vector<string> paths = GetFilePaths(path);
  for (vector<string>::const_iterator physical_path = paths.begin();
       physical_path != paths.end(); ++physical_path) {
    if (::chown(physical_path->c_str(), uid, gid)) {
      return -errno;
    }
  }
  return 0;
}

int SwitchFS::utime(const char* path, struct utimbuf* buffer) {
  vector<string> paths = GetFilePaths(path);
  for (vector<string>::const_iterator physical_path = paths.begin();
       physical_path != paths.end(); ++physical_path) {
    if (::utime(physical_path->c_str(), buffer)) {
      return -errno;
    }
  }
  return 0;
}

int SwitchFS::mkdir(const char* path, mode_t mode) {
  // For each source disk source we have,
  int ret = 0;
  LOG("mkdir mode_t: %d", mode);
  pthread_rwlock_rdlock(&sources_lock_);
  source_map(source) {
    string file_path = Join(*source, path);

    // Create the new directory. Note: If some of these fail while others
    // succeed, we will be left in a bad state.
    if (::mkdir(file_path.c_str(), mode)) {
      ret = -errno;
    }
  }
  pthread_rwlock_unlock(&sources_lock_);
  return 0;
}

int SwitchFS::readdir(const char* path, void* buf,
		      fuse_fill_dir_t filler, off_t offset,
		      struct fuse_file_info* fi) {
  // For each source disk source we have, we keep track of unique files. Note:
  // This implementation could be made fast with a merge sort approach. That
  // would also make sure our output is in the right order which it is not now.
  // Or does another fs layer do a sorting?
  set<string> unique_files;

  pthread_rwlock_rdlock(&sources_lock_);
  source_map(source) {
    string file_path = Join(*source, path);

    DIR* dp = ::opendir(file_path.c_str());
    if (!dp) {
      pthread_rwlock_unlock(&sources_lock_);
      DEBUG("Error: readdir cannot open the dir: " << path << endl);
      return -errno;
    }

    struct dirent *de;
    while ((de = ::readdir(dp))) {
      if (unique_files.find(de->d_name) != unique_files.end())
        continue;
      unique_files.insert(de->d_name);
      if (filler(buf, de->d_name, NULL, 0))
        break;
    }
    ::closedir(dp);
  }
  pthread_rwlock_unlock(&sources_lock_);
  return 0;
}

int SwitchFS::mknod(const char* path, mode_t mode, dev_t dev) {
  // File creation is where we decide which sources (devices) will house this
  // file. The number of sources we store this file on is equal to the desired
  // replication level of the file. The sources we store this file on is decided
  // by the balancing algorithm.
  int replication_level = GetDesiredReplication(path);
  vector<Source*> new_sources = AllocateSources(replication_level);

  DEBUG("Creating new file " << path << " using replication "
	<< replication_level << endl);

  for (vector<Source*>::iterator source = new_sources.begin();
       source != new_sources.end(); ++source) {
    string file_path = Join(**source, path);

    // XXX Note that if some of these fail while others succeed, we will be left
    // in a bad state.
    if (::mknod(file_path.c_str(), mode, dev)) {
      return -errno;
    }
  }
  return 0;
}

int SwitchFS::unlink(const char* path) {
  // We want lazy deletion. This just calls rename with our deleted file prefix
  // in order to move the file(s) there.
  string new_path = Join(SwitchFS::GetDeletedPrefix(), path);

  // We need to ensure the base directories exist.
  stack<string> parents;
  for (string parent_path = StripBasename(new_path);
       parent_path.size() > 0; parent_path = StripBasename(parent_path)) {
    if (SwitchFS::access(parent_path.c_str(), F_OK)) {
      parents.push(parent_path);
    }
  }
  while (!parents.empty()) {
    DEBUG("Creating directory: " << parents.top());
    if (SwitchFS::mkdir(parents.top().c_str(),
			S_IRWXU | S_IRWXG | S_IRWXO)) {
      return -errno;
    }
    parents.pop();
  }

  DEBUG("Unlink redirecting to rename: " << path << " -> " << new_path << endl);

  return SwitchFS::rename(path, new_path.c_str());
}

int SwitchFS::rmdir(const char* virtual_path) {
  pthread_rwlock_rdlock(&sources_lock_);
  source_map(source) {
    string full_path = Join(*source, virtual_path);
    // XXX Note that if some of these fail while others succeed, we will be left
    // in a bad state.
    if (::rmdir(full_path.c_str())) {
      pthread_rwlock_unlock(&sources_lock_);
      return -errno;
    }
  }
  pthread_rwlock_unlock(&sources_lock_);
  return 0;
}

int SwitchFS::rename(const char* old_path, const char* new_path) {
  // Get the locations of this file.
  vector<Source*> sources = GetFileSources(old_path);
  if (sources.size() < 1) {
    DEBUG("Error: rename: file has no sources: " << old_path << endl);
    return -ENOENT;
  }

  // We have to call remove on any existing file we want to overwrite since they
  // may be on different drives and the low level fs doesn't know about this and
  // won't do this for us.
  vector<Source*> targets = GetFileSources(new_path);
  if (targets.size() > 0) {
    DEBUG("rename: backing-up move target: " << new_path << endl);
    if (SwitchFS::unlink(new_path)) {
      return -errno;
    }
  }

  // Perform the renaming in the low level fs.
  for (vector<Source*>::iterator source = sources.begin();
       source != sources.end(); ++source) {
    string full_old_path = Join(**source, old_path);
    string full_new_path = Join(**source, new_path);
    DEBUG("renaming: " << full_old_path << " " << full_new_path << endl);
    if (::rename(full_old_path.c_str(), full_new_path.c_str())) {
      return -errno;
    }
  }
  return 0;
}

int SwitchFS::open(const char *path, struct fuse_file_info *fi) {
  // If we are opening a file for writing, we open and obtain a file descriptor
  // for all real paths associated with this virtual path. This is not necessary
  // for reading, but we do it anyway now for simplicity. The fh member of the
  // fuse_file_info can be used to store a set of open file descriptors.

  set<int>* file_descriptors = new set<int>;
  vector<string> paths = GetFilePaths(path);
  for (vector<string>::const_iterator physical_path = paths.begin();
       physical_path != paths.end(); ++physical_path) {
    // XXX Note that if some of these fail while others succeed, we will be left
    // in a bad state.
    int file_descriptor = ::open(physical_path->c_str(), fi->flags);
    if (file_descriptor == -1) {
      return -errno;
    }
    file_descriptors->insert(file_descriptor);
  }

  // Store this set of file descriptors in the fuse structure.
  fi->fh = reinterpret_cast<uint64_t>(file_descriptors);
  return 0;
}

int SwitchFS::release(const char* path, struct fuse_file_info* fi) {
  // There are no more references to these file descriptors anymore. We can now
  // safely close the files and release the associated memory.
  set<int>* file_descriptors = reinterpret_cast<set<int>*>(fi->fh);
  assert(file_descriptors);
  assert(file_descriptors->size() > 0);

  for (set<int>::const_iterator file_descriptor = file_descriptors->begin();
       file_descriptor != file_descriptors->end(); ++file_descriptor) {
    if (close(*file_descriptor)) {
      return -errno;
    }
  }

  delete file_descriptors;
  fi->fh = 0;

  return 0;
}

int SwitchFS::read(const char *path, char *buf, size_t size,
		   off_t offset, struct fuse_file_info *fi) {
  // The fh member variable of the fuse_file_info stores a pointer to a set of
  // file descriptors associated with this virtual path. For reading operations,
  // we can read from either of them. Here we extract the first from the set and
  // read from it.
  set<int>* file_descriptors = reinterpret_cast<set<int>*>(fi->fh);
  assert(file_descriptors);
  assert(file_descriptors->size() > 0);
  int file_descriptor = *file_descriptors->begin();

  int result = pread(file_descriptor, buf, size, offset);
  if (result == -1) {
    return -errno;
  }
  return result;
}

int SwitchFS::write(const char* path, const char* buffer, size_t size,
		    off_t offset, struct fuse_file_info* fi) {
  // The fh member variable of the fuse_file_info stores a pointer to a set of
  // file descriptors associated with this virtual path. For writing operations,
  // we must write to each one of these. Further optimization of this method
  // could include asynchronous writes here.
  set<int>* file_descriptors = reinterpret_cast<set<int>*>(fi->fh);
  assert(file_descriptors);
  assert(file_descriptors->size() > 0);

  for (set<int>::const_iterator file_descriptor = file_descriptors->begin();
       file_descriptor != file_descriptors->end(); ++file_descriptor) {

    // SwitchFS always reserves space per source for creation of directories.
    // TODO: Determine if the overhead of the call to fstatvfs is measurable.
    struct statvfs source_stat;
    if (fstatvfs(*file_descriptor, &source_stat) == -1) {
      return -errno;
    }
    unsigned long long free_space =
        ((unsigned long long) source_stat.f_bavail) * source_stat.f_bsize;
    unsigned long long bytes_reserved = GetBytesReserved();
    if (free_space < bytes_reserved || size > (free_space - bytes_reserved)) {
      return -ENOSPC;
    }

    // Note that if some of these fail while others succeed, we will be left in
    // a bad state.
    int result = pwrite(*file_descriptor, buffer, size, offset);
    if (result == -1) {
      return -errno;
    }
    // Fuse requires that the amount of bytes specified to this call are
    // written. Since the documentation of pwrite does not necessarily guarantee
    // this, we must check here.
    assert(result == int(size));
  }
  return size;
}

int SwitchFS::fsync(const char* path, int datasync, struct fuse_file_info* fi) {
  set<int>* file_descriptors = reinterpret_cast<set<int>*>(fi->fh);
  assert(file_descriptors);
  assert(file_descriptors->size() > 0);

  for (set<int>::const_iterator file_descriptor = file_descriptors->begin();
       file_descriptor != file_descriptors->end(); ++file_descriptor) {
    if (datasync) {
      if (::fdatasync(*file_descriptor)) {
	return -errno;
      }
    } else {
      if (::fsync(*file_descriptor)) {
	return -errno;
      }
    }
  }
  return 0;
}

int SwitchFS::truncate(const char* path, off_t size) {
  vector<string> paths = GetFilePaths(path);
  for (vector<string>::const_iterator physical_path = paths.begin();
       physical_path != paths.end(); ++physical_path) {
    if (::truncate(physical_path->c_str(), size)) {
      return -errno;
    }
  }
  return 0;
}
